package v2

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"net"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/axgle/mahonia"
	"github.com/gogf/gf/v2/container/garray"
	"github.com/gogf/gf/v2/container/gvar"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gcache"
	"github.com/gogf/gf/v2/os/gcron"
	"github.com/gogf/gf/v2/os/gctx"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/os/gtime"
)

// // 使用全局上下文

var ctx context.Context

// 常量定义
const (
	// 消息标识符
	JT809_HEADER_FLAG_START = 0x5B // 标识位，固定为5B
	JT809_HEADER_FLAG_END   = 0x5D // 标识位，固定为5D

	// 转义字符
	JT809_ESCAPE_CHAR    = 0x5A // 转义字符
	JT809_ESCAPE_CHAR_5B = 0x01 // 0x5B转义为0x5A0x01
	JT809_ESCAPE_CHAR_5A = 0x02 // 0x5A转义为0x5A0x02

	JT809_ESCAPE_PREFIX_5D = 0x5E // 0x5D转义的前缀
	JT809_ESCAPE_PREFIX_5E = 0x5E // 0x5E转义的前缀
	JT809_ESCAPE_SUFFIX_5D = 0x01 // 0x5D转义为0x5E0x01
	JT809_ESCAPE_SUFFIX_5E = 0x02 // 0x5E转义为0x5E0x02

	// 心跳相关
	JT809_HEART_BEAT   = "5B00000018000000001005000C5B510000010000000000F58B5D"
	JT809_CONN_TIMEOUT = 10 * time.Second

	// 消息类型常量
	JT809_MSG_LOGIN              = 0x1001 // 主链路登录请求消息
	JT809_MSG_LOGIN_RESPONSE     = 0x9001 // 主链路登录应答消息
	JT809_MSG_LOGOUT             = 0x1002 // 主链路注销请求消息
	JT809_MSG_LOGOUT_RESPONSE    = 0x9002 // 主链路注销应答消息
	JT809_MSG_HEARTBEAT          = 0x1005 // 主链路心跳请求消息
	JT809_MSG_HEARTBEAT_RESPONSE = 0x9005 // 主链路心跳应答消息
	JT809_MSG_DISCONNECT_INFORM  = 0x9003 // 主链路断开通知消息
)

// JT809Client 客户端结构体
type JT809Client struct {
	conn          net.Conn         // TCP连接
	buffer        *bytes.Buffer    // 数据缓冲区
	mutex         sync.Mutex       // 互斥锁，保护共享资源
	messageBuffer []byte           // 接收到的消息缓冲区
	config        *JT809Config     // 客户端配置
	logger        *glog.Logger     // 日志记录器
	cache         *gcache.Cache    // 缓存
	heartbeatID   *gcron.Entry     // 心跳定时任务Entry
	msgQueue      *garray.StrArray // 消息队列
	isRunning     bool             // 客户端是否在运行
}

// JT809Config 配置结构体
type JT809Config struct {
	ServerIP          string
	ServerPort        int
	UserID            uint32
	Password          string
	DownLinkIP        string
	DownLinkPort      uint16
	MSG_GNSSCENTERID  uint32
	HeartbeatInterval int // 心跳间隔，单位秒
	ReconnectInterval int // 重连间隔，单位秒
}

// JT809Message 消息结构体
type JT809Message struct {
	Header      JT809Header // 消息头
	DataPacket  []byte      // 数据部分
	CRC         uint16      // CRC校验码
	MessageData interface{} // 解析后的业务数据
	Raw         []byte      // 原始消息数据
	ParseTime   time.Time   // 解析时间
	Valid       bool        // 消息是否有效
	Error       error       // 解析过程中的错误
}

// JT809Header 消息头结构体
type JT809Header struct {
	MsgLength       uint32  // 数据长度
	MsgSN           uint32  // 报文序列号
	MsgID           uint16  // 业务数据类型
	MsgGNSSCenterID uint32  // 下级平台接入码
	VersionFlag     [3]byte // 协议版本号
	EncryptFlag     byte    // 报文加密标识
	EncryptKey      uint32  // 数据加密的密钥
}

// SetGlobalContext 设置全局上下文，用于传递main.go中的ctx到本包
func SetGlobalContext(globalCtx context.Context) {
	ctx = globalCtx
}

// 创建新的JT809客户端
func NewJT809Client(ctx context.Context, config *JT809Config) *JT809Client {
	if config == nil {
		config = &JT809Config{
			ServerIP:          "127.0.0.1",
			ServerPort:        8888,
			HeartbeatInterval: 60,
			ReconnectInterval: 10,
		}
	}

	if ctx == nil {
		ctx = gctx.New()
	}

	return &JT809Client{
		buffer:        bytes.NewBuffer(make([]byte, 0, 4096)),
		messageBuffer: make([]byte, 0, 4096),
		config:        config,
		logger:        glog.New(),
		cache:         gcache.New(),
		msgQueue:      garray.NewStrArray(true),
	}
}

// 从配置加载用户信息
func (c *JT809Client) LoadUserFromDB() error {
	res, err := g.DB().Model("fa_jt809Login").Where("id", 1).One()
	if err != nil {
		return err
	}

	c.config.UserID = gvar.New(res["UserID"]).Uint32()
	c.config.Password = gvar.New(res["Password"]).String()
	c.config.DownLinkIP = gvar.New(res["DownLinkIP"]).String()
	c.config.DownLinkPort = gvar.New(res["DownLinkPort"]).Uint16()
	c.config.MSG_GNSSCENTERID = gvar.New(res["MSG_GNSSCENTERID"]).Uint32()

	return nil
}

// 连接服务器
func (c *JT809Client) Connect() error {
	addr := fmt.Sprintf("%s:%d", c.config.ServerIP, c.config.ServerPort)
	conn, err := net.DialTimeout("tcp", addr, JT809_CONN_TIMEOUT)
	if err != nil {
		return fmt.Errorf("连接服务器失败: %v", err)
	}

	c.conn = conn
	return nil
}

// 关闭连接
func (c *JT809Client) Close() error {
	if c.conn != nil {
		return c.conn.Close()
	}
	return nil
}

// 构建登录消息 - 使用与v1版本完全一致的方式
func (c *JT809Client) BuildLoginMessage() ([]byte, error) {
	// 从数据库获取用户信息
	if err := c.LoadUserFromDB(); err != nil {
		return nil, fmt.Errorf("加载用户信息失败: %v", err)
	}
	// 直接使用模拟v1版本的方式，确保结果完全一致
	loginMsg, v1MsgHex := createV1StyleLoginMessage(context.Background(), c.logger, c.config)

	// 输出原始v1风格的消息内容
	c.logger.Info(context.Background(), "构建的登录消息(v1风格):", v1MsgHex)

	return loginMsg, nil
}

// 直接使用v1的方式创建登录消息，返回字节和十六进制字符串
func createV1StyleLoginMessage(ctx context.Context, logger *glog.Logger, config *JT809Config) ([]byte, string) {
	// 初始化基本字段，完全按照v1的方式
	msgLength := "00000000"                                                  // 临时值，后面计算
	msgSN := strconv.FormatInt(int64(0), 16)                                 // 序列号为0
	msgID := "1001"                                                          // 消息ID固定为0x1001
	msgGNSSCenterID := strconv.FormatInt(int64(config.MSG_GNSSCENTERID), 16) // GNSS中心ID
	versionFlag := "01020F"                                                  // 版本号1.2.15
	encryptFlag := "00"                                                      // 不加密
	encryptKey := strconv.FormatInt(int64(0), 16)                            // 加密密钥为0
	userID := strconv.FormatInt(int64(config.UserID), 16)                    // 用户ID
	password := hex.EncodeToString([]byte(config.Password))                  // 密码
	downLinkIP := hex.EncodeToString([]byte(config.DownLinkIP))              // 下行链路IP
	downLinkPort := strconv.FormatInt(int64(config.DownLinkPort), 16)        // 下行链路端口
	crcCode := strconv.FormatInt(int64(9578), 16)                            // CRC固定为9578

	// 字段长度标准化
	// 1. MSG_SN
	if len(msgSN) < 8 {
		supple := 8 - len(msgSN)
		msgSN = supplementZeros(supple) + msgSN
	}

	// 2. MSG_ID
	if len(msgID) < 4 {
		supple := 4 - len(msgID)
		msgID = supplementZeros(supple) + msgID
	}

	// 3. MSG_GNSSCENTERID
	if len(msgGNSSCenterID) < 8 {
		supple := 8 - len(msgGNSSCenterID)
		msgGNSSCenterID = supplementZeros(supple) + msgGNSSCenterID
	}

	// 4. ENCRYPT_KEY
	if len(encryptKey) < 8 {
		supple := 8 - len(encryptKey)
		encryptKey = supplementZeros(supple) + encryptKey
	}

	// 5. UserID
	if len(userID) < 8 {
		supple := 8 - len(userID)
		userID = supplementZeros(supple) + userID
	}

	// 6. Password
	if len(password) < 16 {
		supple := 16 - len(password)
		password = supplementZeros(supple) + password
	}

	// 7. DownLinkIP
	if len(downLinkIP) < 64 {
		supple := 64 - len(downLinkIP)
		downLinkIP = downLinkIP + supplementZeros(supple)
	}

	// 8. DownLinkPort
	if len(downLinkPort) < 4 {
		supple := 4 - len(downLinkPort)
		downLinkPort = supplementZeros(supple) + downLinkPort
	}

	// 打印每个字段的值，调试用
	logger.Info(ctx, "消息字段(转义前):")
	logger.Info(ctx, "  msgSN: ", msgSN)
	logger.Info(ctx, "  msgID: ", msgID)
	logger.Info(ctx, "  msgGNSSCenterID: ", msgGNSSCenterID)
	logger.Info(ctx, "  versionFlag: ", versionFlag)
	logger.Info(ctx, "  encryptFlag: ", encryptFlag)
	logger.Info(ctx, "  encryptKey: ", encryptKey)
	logger.Info(ctx, "  userID: ", userID)
	logger.Info(ctx, "  password: ", password)
	logger.Info(ctx, "  downLinkIP: ", downLinkIP)
	logger.Info(ctx, "  downLinkPort: ", downLinkPort)
	logger.Info(ctx, "  crcCode: ", crcCode)

	// 拼接所有字段计算长度
	str := msgLength + msgSN + msgID + msgGNSSCenterID + versionFlag + encryptFlag +
		encryptKey + userID + password + downLinkIP + downLinkPort + crcCode

	// 计算并设置消息长度
	lenth := strconv.FormatInt(int64(len(str)/2), 16)
	if len(lenth) < 8 {
		supple := 8 - len(lenth)
		msgLength = supplementZeros(supple) + lenth
	}

	// 重新拼接消息头
	str = msgLength + msgSN + msgID + msgGNSSCenterID + versionFlag + encryptFlag +
		encryptKey + userID + password + downLinkIP + downLinkPort + crcCode

	// 打印最终的消息字符串(转义前)
	logger.Info(ctx, "最终消息字符串(转义前): ", str)
	logger.Info(ctx, "消息长度(转义前): ", len(str)/2)

	// 添加标识符并转义处理，完全按照v1的方式
	finalMsg := "5b" + makeOriginalV1Message(str) + "5d"

	// 转换为字节
	msgBytes, _ := hex.DecodeString(finalMsg)

	return msgBytes, finalMsg
}

// 完全按照v1的方式处理消息
func makeOriginalV1Message(str string) string {
	save := []string{}
	for i, v := range str {
		if i%2 != 0 {
			save = append(save, string(str[i-1])+string(v))
		}
	}
	// 使用v1原始的转义方式
	return originalV1Escape(save)
}

// 原始的v1转义处理
func originalV1Escape(arr []string) string {
	for k, v := range arr {
		if v == "5b" {
			arr[k] = "5a01"
		} else if v == "5a" {
			arr[k] = "5a02"
		} else if v == "5d" {
			arr[k] = "5e01"
		} else if v == "5e" {
			arr[k] = "5e02"
		}
	}
	strArrayNew := strings.Join(arr, "")
	return strArrayNew
}

// 获取v1版本的登录消息
func getV1LoginMessage(ctx context.Context, logger *glog.Logger) {
	// 从数据库获取用户信息
	res, err := g.DB().Model("fa_jt809Login").Where("id", 1).One()
	if err != nil {
		logger.Error(ctx, "获取V1登录消息失败(数据库错误):", err)
		return
	}

	// 创建一个临时配置
	tempConfig := &JT809Config{
		UserID:           gvar.New(res["UserID"]).Uint32(),
		Password:         gvar.New(res["Password"]).String(),
		DownLinkIP:       gvar.New(res["DownLinkIP"]).String(),
		DownLinkPort:     gvar.New(res["DownLinkPort"]).Uint16(),
		MSG_GNSSCENTERID: gvar.New(res["MSG_GNSSCENTERID"]).Uint32(),
	}

	// 调用真实的v1代码，通过模拟执行流程
	v1LoginMsg := mockV1LoginProcess(ctx, logger, tempConfig)
	logger.Info(ctx, "真实V1登录消息(hex):", v1LoginMsg)
}

// 模拟v1的登录消息创建过程
func mockV1LoginProcess(ctx context.Context, logger *glog.Logger, config *JT809Config) string {
	// 1. 创建初始消息结构
	var loginSend struct {
		MSG_LENGTH       string
		MSG_SN           string
		MSG_ID           string
		MSG_GNSSCENTERID string
		VERSION_FLAG     string
		ENCRYPT_FLAG     string
		ENCRYPT_KEY      string
		UserID           string
		Password         string
		DownLinkIP       string
		DownLinkPort     string
		CRC_CODE         string
	}

	// 2. 设置初始值
	loginSend.MSG_LENGTH = "00000000"
	loginSend.MSG_SN = strconv.FormatInt(int64(0), 16)
	loginSend.MSG_ID = "1001"
	loginSend.MSG_GNSSCENTERID = strconv.FormatInt(int64(config.MSG_GNSSCENTERID), 16)
	loginSend.VERSION_FLAG = "01020F"
	loginSend.ENCRYPT_FLAG = "00"
	loginSend.ENCRYPT_KEY = strconv.FormatInt(int64(0), 16)
	loginSend.UserID = strconv.FormatInt(int64(config.UserID), 16)
	loginSend.Password = hex.EncodeToString([]byte(config.Password))
	loginSend.DownLinkIP = hex.EncodeToString([]byte(config.DownLinkIP))
	loginSend.DownLinkPort = strconv.FormatInt(int64(config.DownLinkPort), 16)
	loginSend.CRC_CODE = strconv.FormatInt(int64(9578), 16)

	// 3. 应用长度检查和填充
	// 3.1 MSG_SN
	if gvar.New(loginSend.MSG_SN).Int() < 8 {
		supple := 8 - len(loginSend.MSG_SN)
		loginSend.MSG_SN = supplementZeros(supple) + loginSend.MSG_SN
	}

	// 3.2 MSG_ID
	if gvar.New(loginSend.MSG_ID).Int() < 4 {
		supple := 4 - len(loginSend.MSG_ID)
		loginSend.MSG_ID = supplementZeros(supple) + loginSend.MSG_ID
	}

	// 3.3 MSG_GNSSCENTERID
	if gvar.New(loginSend.MSG_GNSSCENTERID).Int() < 8 {
		supple := 8 - len(loginSend.MSG_GNSSCENTERID)
		loginSend.MSG_GNSSCENTERID = supplementZeros(supple) + loginSend.MSG_GNSSCENTERID
	}

	// 3.4 ENCRYPT_KEY
	if gvar.New(loginSend.ENCRYPT_KEY).Int() < 8 {
		supple := 8 - len(loginSend.ENCRYPT_KEY)
		loginSend.ENCRYPT_KEY = supplementZeros(supple) + loginSend.ENCRYPT_KEY
	}

	// 3.5 UserID
	if gvar.New(loginSend.UserID).Int() < 8 {
		supple := 8 - len(loginSend.UserID)
		loginSend.UserID = supplementZeros(supple) + loginSend.UserID
	}

	// 3.6 Password
	if gvar.New(loginSend.Password).Int() < 16 {
		supple := 16 - len(loginSend.Password)
		loginSend.Password = supplementZeros(supple) + loginSend.Password
	}

	// 3.7 DownLinkIP
	if gvar.New(loginSend.DownLinkIP).Int() < 64 {
		supple := 64 - len(loginSend.DownLinkIP)
		loginSend.DownLinkIP = loginSend.DownLinkIP + supplementZeros(supple)
	}

	// 3.8 DownLinkPort
	if gvar.New(loginSend.DownLinkPort).Int() < 4 {
		supple := 4 - len(loginSend.DownLinkPort)
		loginSend.DownLinkPort = supplementZeros(supple) + loginSend.DownLinkPort
	}

	// 4. 拼接所有字段计算长度
	str := loginSend.MSG_LENGTH + loginSend.MSG_SN + loginSend.MSG_ID +
		loginSend.MSG_GNSSCENTERID + loginSend.VERSION_FLAG + loginSend.ENCRYPT_FLAG +
		loginSend.ENCRYPT_KEY + loginSend.UserID + loginSend.Password +
		loginSend.DownLinkIP + loginSend.DownLinkPort + loginSend.CRC_CODE

	lenth := gvar.New(strconv.FormatInt(int64(len(str)/2), 16)).String()
	if len(lenth) < 8 {
		supple := 8 - len(lenth)
		loginSend.MSG_LENGTH = supplementZeros(supple) + lenth
	}

	// 5. 重新拼接消息
	str = loginSend.MSG_LENGTH + loginSend.MSG_SN + loginSend.MSG_ID +
		loginSend.MSG_GNSSCENTERID + loginSend.VERSION_FLAG + loginSend.ENCRYPT_FLAG +
		loginSend.ENCRYPT_KEY + loginSend.UserID + loginSend.Password +
		loginSend.DownLinkIP + loginSend.DownLinkPort + loginSend.CRC_CODE

	// 6. 打印详细信息
	logger.Info(ctx, "V1模拟各字段值:")
	logger.Info(ctx, "  MSG_LENGTH:", loginSend.MSG_LENGTH)
	logger.Info(ctx, "  MSG_SN:", loginSend.MSG_SN)
	logger.Info(ctx, "  MSG_ID:", loginSend.MSG_ID)
	logger.Info(ctx, "  MSG_GNSSCENTERID:", loginSend.MSG_GNSSCENTERID)
	logger.Info(ctx, "  VERSION_FLAG:", loginSend.VERSION_FLAG)
	logger.Info(ctx, "  ENCRYPT_FLAG:", loginSend.ENCRYPT_FLAG)
	logger.Info(ctx, "  ENCRYPT_KEY:", loginSend.ENCRYPT_KEY)
	logger.Info(ctx, "  UserID:", loginSend.UserID)
	logger.Info(ctx, "  Password:", loginSend.Password)
	logger.Info(ctx, "  DownLinkIP:", loginSend.DownLinkIP)
	logger.Info(ctx, "  DownLinkPort:", loginSend.DownLinkPort)
	logger.Info(ctx, "  CRC_CODE:", loginSend.CRC_CODE)
	logger.Info(ctx, "  Original str:", str)

	// 7. 添加标识符并进行转义处理
	splitChars := []string{}
	for i, v := range str {
		if i%2 != 0 {
			splitChars = append(splitChars, string(str[i-1])+string(v))
		}
	}

	for k, v := range splitChars {
		if v == "5b" {
			splitChars[k] = "5a01"
		} else if v == "5a" {
			splitChars[k] = "5a02"
		} else if v == "5d" {
			splitChars[k] = "5e01"
		} else if v == "5e" {
			splitChars[k] = "5e02"
		}
	}

	processedStr := strings.Join(splitChars, "")
	finalMsg := "5b" + processedStr + "5d"

	return finalMsg
}

// 补充指定数量的0
func supplementZeros(num int) string {
	var zeros string
	for i := 0; i < num; i++ {
		zeros += "0"
	}
	return zeros
}

// 与v1.Make方法一致的实现
func makeJT809Msg(str string) string {
	save := []string{}
	for i, v := range str {
		if i%2 != 0 {
			save = append(save, string(str[i-1])+string(v))
		}
	}
	return v1StyleEscape(save)
}

// 与v1.ortherStartZhuanyi方法一致的实现
func v1StyleEscape(arr []string) string {
	for k, v := range arr {
		if strings.Contains(v, "5b") {
			arr[k] = strings.Replace(v, "5b", "5a01", 1)
		}
		if strings.Contains(v, "5a") {
			arr[k] = strings.Replace(v, "5a", "5a02", 1)
		}
		if strings.Contains(v, "5d") {
			arr[k] = strings.Replace(v, "5d", "5e01", 1)
		}
		if strings.Contains(v, "5e") {
			arr[k] = strings.Replace(v, "5e", "5e02", 1)
		}
	}
	strArrayNew := strings.Join(arr, ",")
	return strings.Replace(strArrayNew, ",", "", -1)
}

// 发送消息
func (c *JT809Client) Send(data []byte) error {
	if c.conn == nil {
		return errors.New("连接未建立")
	}

	c.mutex.Lock()
	defer c.mutex.Unlock()

	_, err := c.conn.Write(data)
	return err
}

// 发送心跳
func (c *JT809Client) SendHeartbeat() error {
	heartbeatBytes, err := hex.DecodeString(JT809_HEART_BEAT)
	if err != nil {
		return err
	}
	return c.Send(heartbeatBytes)
}

// 接收消息
func (c *JT809Client) Receive() ([]byte, error) {
	if c.conn == nil {
		return nil, errors.New("连接未建立")
	}

	// 设置读取超时，防止永久阻塞
	if err := c.conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil {
		return nil, fmt.Errorf("设置读取超时失败: %v", err)
	}

	buffer := make([]byte, 4096)
	n, err := c.conn.Read(buffer)
	if err != nil {
		// 区分超时和其他错误
		if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
			return nil, nil // 返回nil表示超时，但不是错误
		}
		return nil, fmt.Errorf("读取数据失败: %v", err)
	}

	if n == 0 {
		return nil, nil // 无数据
	}

	c.mutex.Lock()
	defer c.mutex.Unlock()

	// 将新收到的数据追加到缓冲区
	c.messageBuffer = append(c.messageBuffer, buffer[:n]...)

	// 尝试解析完整消息
	return c.parseMessages()
}

// 解析缓冲区中的完整消息，处理粘包和半包问题
func (c *JT809Client) parseMessages() ([]byte, error) {
	var startIdx, endIdx = -1, -1

	// 查找消息的起始和结束标记
	for i := 0; i < len(c.messageBuffer)-1; i++ {
		if c.messageBuffer[i] == JT809_HEADER_FLAG_START {
			startIdx = i
		} else if c.messageBuffer[i] == JT809_HEADER_FLAG_END && startIdx != -1 {
			endIdx = i
			break
		}
	}

	// 如果没有找到完整消息，返回nil
	if startIdx == -1 || endIdx == -1 {
		return nil, nil
	}

	// 提取完整消息
	completeMessage := c.messageBuffer[startIdx : endIdx+1]

	// 更新缓冲区，移除已处理的消息
	if endIdx+1 < len(c.messageBuffer) {
		c.messageBuffer = c.messageBuffer[endIdx+1:]
	} else {
		c.messageBuffer = c.messageBuffer[:0]
	}

	// 返回提取的消息
	return completeMessage, nil
}

// 解析JT809消息
func (c *JT809Client) ParseMessage(data []byte) (*JT809Message, error) {
	// 创建消息结构，并设置默认值
	msg := &JT809Message{
		Raw:       data,
		ParseTime: time.Now(),
		Valid:     false,
	}

	// 检查消息是否完整
	if len(data) < 2 || data[0] != JT809_HEADER_FLAG_START || data[len(data)-1] != JT809_HEADER_FLAG_END {
		msg.Error = errors.New("无效的消息格式")
		return msg, msg.Error
	}

	// 反转义
	unescapedData := unescapeMessage(data)

	if len(unescapedData) < 26 { // 消息头(22) + CRC(2) + 开始/结束标识(2)
		msg.Error = errors.New("消息长度不足")
		return msg, msg.Error
	}

	// 提取消息内容（去掉开始和结束标记）
	content := unescapedData[1 : len(unescapedData)-1]

	// 解析消息头
	reader := bytes.NewReader(content[:22])

	var header JT809Header
	if err := binary.Read(reader, binary.BigEndian, &header); err != nil {
		msg.Error = fmt.Errorf("解析消息头失败: %w", err)
		return msg, msg.Error
	}

	// 设置消息头
	msg.Header = header

	// 检查消息体长度
	if header.MsgLength != uint32(len(content)+2) {
		glog.Warning(ctx, "消息长度不匹配, 头部声明:", header.MsgLength, "实际长度:", len(content)+2)
		// 继续处理，不返回错误
	}

	// 提取数据部分
	if len(content) > 24 { // 确保数据长度足够
		msg.DataPacket = content[22 : len(content)-2]
	} else {
		msg.DataPacket = []byte{}
	}

	// 提取CRC
	if len(content) >= 2 {
		msg.CRC = binary.BigEndian.Uint16(content[len(content)-2:])
	}

	// 验证CRC
	calculatedCRC := calculateCRC(content[:len(content)-2])
	if calculatedCRC != msg.CRC {
		msg.Error = fmt.Errorf("CRC校验失败: 期望值=%d, 计算值=%d", msg.CRC, calculatedCRC)
		glog.Warning(ctx, msg.Error)
		// 继续处理，不返回错误
	}

	// 标记消息有效
	msg.Valid = true

	// 根据消息类型解析数据包（可以在这里添加具体类型的解析）
	switch msg.Header.MsgID {
	case 0x1001: // 登录应答
		// TODO: 实现登录应答消息的解析
	case 0x1005: // 心跳应答
		// TODO: 实现心跳应答消息的解析
	default:
		// 其他类型消息暂不解析
	}

	return msg, nil
}

// 获取消息类型名称
func getMsgTypeName(msgID uint16) string {
	switch msgID {
	case JT809_MSG_LOGIN:
		return "主链路登录请求"
	case JT809_MSG_LOGIN_RESPONSE:
		return "主链路登录应答"
	case JT809_MSG_LOGOUT:
		return "主链路注销请求"
	case JT809_MSG_LOGOUT_RESPONSE:
		return "主链路注销应答"
	case JT809_MSG_HEARTBEAT:
		return "主链路心跳请求"
	case JT809_MSG_HEARTBEAT_RESPONSE:
		return "主链路心跳应答"
	case JT809_MSG_DISCONNECT_INFORM:
		return "主链路断开通知"
	default:
		return fmt.Sprintf("未知消息类型(0x%04X)", msgID)
	}
}

// 处理收到的消息
func (c *JT809Client) handleReceivedData(data []byte) {
	// 将收到的二进制数据转为十六进制字符串
	hexStr := hex.EncodeToString(data)

	// 推送到消息队列
	c.msgQueue.PushRight(hexStr)

	// 解析消息（可选）
	msg, err := c.ParseMessage(data)
	if err != nil {
		glog.Error(ctx, "解析消息失败:", err)
		return
	}

	// 根据消息类型处理
	glog.Info(ctx, "收到消息:", getMsgTypeName(msg.Header.MsgID))

	switch msg.Header.MsgID {
	case JT809_MSG_LOGIN_RESPONSE: // 登录应答
		c.handleLoginResponse(msg)
	case JT809_MSG_HEARTBEAT_RESPONSE: // 心跳应答
		c.handleHeartbeatResponse(msg)
	case JT809_MSG_DISCONNECT_INFORM: // 断开通知
		c.handleDisconnectInform(msg)
	default:
		glog.Info(ctx, "收到未处理的消息类型:", getMsgTypeName(msg.Header.MsgID))
	}
}

// 处理登录应答
func (c *JT809Client) handleLoginResponse(msg *JT809Message) {
	if len(msg.DataPacket) < 1 {
		glog.Error(ctx, "登录应答数据包内容不足")
		return
	}

	result := msg.DataPacket[0]
	switch result {
	case 0x00:
		glog.Info(ctx, "登录成功")
	case 0x01:
		glog.Error(ctx, "IP地址不正确")
	case 0x02:
		glog.Error(ctx, "接入码不正确")
	case 0x03:
		glog.Error(ctx, "用户没有注册")
	case 0x04:
		glog.Error(ctx, "密码错误")
	case 0x05:
		glog.Error(ctx, "资源紧张，稍后再连接")
	case 0x06:
		glog.Error(ctx, "其他错误")
	default:
		glog.Error(ctx, "未知的登录结果码:", result)
	}
}

// 处理心跳应答
func (c *JT809Client) handleHeartbeatResponse(msg *JT809Message) {
	glog.Debug(ctx, "收到心跳应答")
}

// 处理断开通知
func (c *JT809Client) handleDisconnectInform(msg *JT809Message) {
	glog.Warning(ctx, "服务器通知断开连接")

	// 可以在这里进行相应的处理，比如重连
	go func() {
		if err := c.Reconnect(); err != nil {
			glog.Error(ctx, "重连失败:", err)
		}
	}()
}

// 启动JT809客户端
func (c *JT809Client) Start() error {
	// 设置运行状态
	c.isRunning = true
	defer func() {
		c.isRunning = false
	}()

	// 使用可取消的上下文
	ctx, cancel := context.WithCancel(gctx.New())
	defer cancel()

	glog.Info(ctx, "正在启动JT809客户端...")

	// 加载配置
	if err := c.LoadUserFromDB(); err != nil {
		return fmt.Errorf("加载用户配置失败: %w", err)
	}
	glog.Info(ctx, "配置已加载")

	// 连接服务器
	if err := c.Connect(); err != nil {
		return fmt.Errorf("连接服务器失败: %w", err)
	}
	glog.Info(ctx, "已连接到服务器")

	// 使用defer确保在函数退出时关闭连接
	defer func() {
		if err := c.Close(); err != nil {
			glog.Error(ctx, "关闭连接失败:", err)
		}
	}()

	// 构建并发送登录消息
	loginMsg, err := c.BuildLoginMessage()
	if err != nil {
		return fmt.Errorf("构建登录消息失败: %w", err)
	}

	glog.Info(ctx, "登录消息构建成功:", hex.EncodeToString(loginMsg))

	if err := c.Send(loginMsg); err != nil {
		return fmt.Errorf("发送登录消息失败: %w", err)
	}
	glog.Info(ctx, "登录消息已发送")

	// 启动心跳定时任务
	heartbeatCtx, heartbeatCancel := context.WithCancel(ctx)
	heartbeatPattern := fmt.Sprintf("*/%d * * * * *", c.config.HeartbeatInterval)

	heartbeatID, err := gcron.Add(heartbeatCtx, heartbeatPattern, func(ctx context.Context) {
		if err := c.SendHeartbeat(); err != nil {
			glog.Error(ctx, "发送心跳失败:", err)
		} else {
			glog.Debug(ctx, "心跳发送成功:", gtime.Now().Format("Y-m-d H:i:s"))
		}
	})

	if err != nil {
		heartbeatCancel()
		return fmt.Errorf("启动心跳任务失败: %w", err)
	}

	// 保存心跳任务ID
	c.heartbeatID = heartbeatID
	glog.Info(ctx, "心跳任务已启动, ID:", heartbeatID)

	// 在函数退出时清理资源
	defer func() {
		if c.heartbeatID != nil {
			c.heartbeatID.Close()
			c.heartbeatID = nil
		}
	}()

	// 启动消息接收和处理循环
	receiveCtx, receiveCancel := context.WithCancel(ctx)
	go func() {
		defer receiveCancel()

		for {
			select {
			case <-receiveCtx.Done():
				glog.Info(ctx, "消息接收循环已停止")
				return
			default:
				if c.conn == nil {
					glog.Error(ctx, "连接已断开")
					return
				}

				data, err := c.Receive()
				if err != nil {
					glog.Error(ctx, "接收消息失败:", err)
					return
				}

				if data != nil {
					glog.Debug(ctx, "收到数据:", hex.EncodeToString(data))
					c.handleReceivedData(data)
				} else {
					// 短暂休眠，避免CPU占用过高
					time.Sleep(100 * time.Millisecond)
				}
			}
		}
	}()

	// 启动消息处理队列
	processCtx, processCancel := context.WithCancel(ctx)
	go func() {
		defer processCancel()
		// 使用processCtx来控制消息处理循环
		for {
			select {
			case <-processCtx.Done():
				glog.Info(ctx, "消息处理队列已停止")
				return
			default:
				c.processMessageQueue()
				// 短暂休眠，避免CPU占用过高
				time.Sleep(100 * time.Millisecond)
			}
		}
	}()

	// 等待上下文取消
	<-ctx.Done()

	glog.Info(ctx, "JT809客户端已停止")
	return nil
}

// 重新连接
func (c *JT809Client) Reconnect() error {
	// 关闭旧连接
	if c.conn != nil {
		c.conn.Close()
		c.conn = nil
	}

	// 停止心跳任务
	if c.heartbeatID != nil {
		c.heartbeatID.Close()
		c.heartbeatID = nil
	}

	// 清空缓冲区
	c.mutex.Lock()
	c.messageBuffer = c.messageBuffer[:0]
	c.mutex.Unlock()

	// 重新启动
	return c.Start()
}

// 处理消息队列
func (c *JT809Client) processMessageQueue() {
	for {
		value, ok := c.msgQueue.PopLeft()
		if ok && value != "" {
			// 直接实现消息处理，避免依赖v1.decode的私有函数
			processMessage(value)
		} else {
			// 队列为空，休眠一段时间
			time.Sleep(100 * time.Millisecond)
		}
	}
}

// 处理接收到的消息，实现与v1.decode类似的功能
func processMessage(strDecode string) {
	// 检查消息格式是否正确（以5b开头，5d结尾）
	if len(strDecode) < 4 || strDecode[0:2] != "5b" || strDecode[len(strDecode)-2:] != "5d" {
		glog.Warning(ctx, "无效的消息格式:", strDecode)
		return
	}

	// 提取消息主体（去掉开头的5b和结尾的5d）
	info3 := strDecode[2 : len(strDecode)-2]
	info3 = processEscapeChars(info3)

	// 检查消息长度
	if len(info3) < 100 {
		glog.Warning(ctx, "消息长度不足:", strDecode)
		return
	}

	// 处理多条消息的情况（粘包）
	if strings.Contains(strDecode, "5d5b00") {
		strDecodeNew := strings.Replace(strDecode, "5d5b00", "5d,5b00", -1)
		strArray := strings.Split(strDecodeNew, ",")
		for _, v := range strArray {
			if len(v) < 4 {
				continue
			}

			msgContent := v[2 : len(v)-2]
			msgContent = processEscapeChars(msgContent)

			// 验证消息长度
			if !isValidMessageLength(msgContent) {
				glog.Warning(ctx, "消息长度校验失败:", v)
				saveUnknownMessage(v, msgContent)
				continue
			}

			// 保存消息
			saveMessage(msgContent)
		}
		return
	}

	// 单条消息处理
	if !isValidMessageLength(info3) {
		glog.Warning(ctx, "消息长度校验失败:", strDecode)
		saveUnknownMessage(strDecode, info3)
		return
	}

	// 保存消息
	saveMessage(info3)
}

// 处理转义字符
func processEscapeChars(str string) string {
	save := []string{}

	// 将十六进制字符串拆分为两字符一组
	for i, v := range str {
		if i%2 != 0 {
			save = append(save, string(str[i-1])+string(v))
		}
	}

	// 处理转义字符
	strArrayNew := strings.Join(save, ",")
	strArrayNew = processSpecialChars(strArrayNew)
	return strArrayNew
}

// 处理特殊字符的转义
func processSpecialChars(str string) string {
	// 处理5a01 -> 5b
	if strings.Contains(str, "5a,01") {
		str = strings.Replace(str, "5a,01", "&#1", -1)
	}

	// 处理5a02 -> 5a
	if strings.Contains(str, "5a,02") {
		str = strings.Replace(str, "5a,02", "&#2", -1)
	}

	// 处理5e01 -> 5d
	if strings.Contains(str, "5e,01") {
		str = strings.Replace(str, "5e,01", "&#3", -1)
	}

	// 处理5e02 -> 5e
	if strings.Contains(str, "5e,02") {
		str = strings.Replace(str, "5e,02", "&#4", -1)
	}

	// 还原特殊字符
	str = strings.Replace(str, "&#1", "5b", -1)
	str = strings.Replace(str, "&#2", "5a", -1)
	str = strings.Replace(str, "&#3", "5d", -1)
	str = strings.Replace(str, "&#4", "5e", -1)

	// 移除逗号
	str = strings.Replace(str, ",", "", -1)
	return str
}

// 检查消息长度是否有效
func isValidMessageLength(info3 string) bool {
	if len(info3) < 8 {
		return false
	}

	// 提取消息长度字段
	msgLengthHex := info3[0:8]
	msgLength, err := strconv.ParseInt(msgLengthHex, 16, 64)
	if err != nil {
		return false
	}

	// 验证消息长度是否与实际长度匹配
	return (msgLength * 2) == int64(len(info3)+4) // +4 是因为5b和5d各占2字节
}

// 保存未知格式的消息到数据库
func saveUnknownMessage(raw, content string) {
	_, err := g.DB().Model("fa_unknow").Data(g.Map{
		"text": raw,
		"time": gtime.Timestamp(),
		"rank": content,
		"len":  len(content) + 4,
	}).Insert()

	if err != nil {
		glog.Error(ctx, "保存未知消息失败:", err)
	}
}

// 保存消息到数据库
func saveMessage(info3 string) {
	// 解析消息头（用于日志或调试，不再使用headerData变量避免警告）
	parseMessageHeader(info3)

	// 解析数据部分
	data := parseMessageData(info3, 100) // 数据部分从100字节开始

	// 调试输出数据
	glog.Info(ctx, "保存数据内容:", data)

	// 保存到数据库
	_, err := g.DB().Model("fa_xjtcp").OmitEmpty().Data(data).Save()
	if err == nil {
		return
	}

	glog.Error(ctx, "保存消息失败:", err)

	// 检查数据库字段编码问题
	if !strings.Contains(err.Error(), "Incorrect string value") {
		return
	}

	glog.Error(ctx, "编码错误，请确保数据库字段支持UTF-8编码")

	// 尝试创建临时表存储数据
	_, execErr := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS fa_xjtcp_temp LIKE fa_xjtcp;`)
	if execErr != nil {
		glog.Error(ctx, "创建临时表失败:", execErr)
		return
	}

	_, execErr = g.DB().Exec(ctx, `ALTER TABLE fa_xjtcp_temp CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;`)
	if execErr != nil {
		glog.Error(ctx, "修改临时表字符集失败:", execErr)
		return
	}

	_, insertErr := g.DB().Model("fa_xjtcp_temp").OmitEmpty().Data(data).Insert()
	if insertErr != nil {
		glog.Error(ctx, "保存到临时表也失败:", insertErr)
		return
	}

	glog.Info(ctx, "数据已保存到临时表fa_xjtcp_temp")
}

// 解析消息头
func parseMessageHeader(info3 string) map[string]interface{} {
	result := make(map[string]interface{})

	// 检查信息长度是否足够
	if len(info3) < 100 {
		return result
	}

	// 解析基本字段，添加错误处理
	msgLength, err := strconv.ParseInt(info3[0:8], 16, 64)
	if err != nil {
		msgLength = 0
	}

	msgSN, err := strconv.ParseInt(info3[8:16], 16, 64)
	if err != nil {
		msgSN = 0
	}

	msgID := info3[16:20]

	msgGNSSCenterID, err := strconv.ParseInt(info3[20:28], 16, 64)
	if err != nil {
		msgGNSSCenterID = 0
	}

	// 版本号
	versionFlag1, err := strconv.ParseInt(info3[28:30], 16, 64)
	if err != nil {
		versionFlag1 = 0
	}

	versionFlag2, err := strconv.ParseInt(info3[30:32], 16, 64)
	if err != nil {
		versionFlag2 = 0
	}

	versionFlag3, err := strconv.ParseInt(info3[32:34], 16, 64)
	if err != nil {
		versionFlag3 = 0
	}

	versionFlag := fmt.Sprintf("%d.%d.%d", versionFlag1, versionFlag2, versionFlag3)

	// 加密标识
	encryptFlag, err := strconv.ParseInt(info3[34:36], 16, 64)
	if err != nil {
		encryptFlag = 0
	}

	encryptKey, err := strconv.ParseInt(info3[36:44], 16, 64)
	if err != nil {
		encryptKey = 0
	}

	// 车牌号和颜色
	var vehicleNo []byte
	if len(info3) >= 86 {
		vehicleNo, _ = hex.DecodeString(info3[44:86])
	}

	vehicleColor := ""
	if len(info3) >= 88 {
		vehicleColor = info3[86:88]
	}

	// 子业务类型
	dataType := ""
	if len(info3) >= 92 {
		dataType = info3[88:92]
	}

	dataLength := int64(0)
	if len(info3) >= 100 {
		dataLength, _ = strconv.ParseInt(info3[92:100], 16, 64)
	}

	// 填充结果
	result["MSG_LENGTH"] = msgLength
	result["MSG_SN"] = msgSN
	result["MSG_ID"] = msgID
	result["MSG_GNSSCENTERID"] = msgGNSSCenterID
	result["VERSION_FLAG"] = versionFlag
	result["ENCRYPT_FLAG"] = encryptFlag
	result["ENCRYPT_KEY"] = encryptKey
	result["VEHICLE_NO"] = string(vehicleNo)
	result["VEHICLE_COLOR"] = vehicleColor
	result["DATA_TYPE"] = dataType
	result["DATA_LENGTH"] = dataLength

	return result
}

// 解析消息数据部分
func parseMessageData(info3 string, startPos int) map[string]interface{} {
	result := make(map[string]interface{})

	// 获取字段映射配置
	res, err := g.DB().Model("fa_daoru2").All()
	if err != nil {
		glog.Error(ctx, "获取字段配置失败:", err)
		return result
	}

	fields := gvar.New(res).Maps()
	position := startPos

	// 根据配置解析各字段
	for _, field := range fields {
		byteSize := gvar.New(field["byte"]).Int()
		if position+(2*byteSize) > len(info3) {
			break
		}

		// 提取字段值
		fieldHex := info3[position : position+(2*byteSize)]
		fieldBytes, _ := hex.DecodeString(fieldHex)

		// 处理GBK编码（如果需要）
		fieldValue := convertGBKToUTF8(string(fieldBytes))

		// 保存字段
		fieldName := gvar.New(field["map"]).String()
		result[fieldName] = removeNullBytes(fieldValue)

		// 更新位置
		position += (2 * byteSize)
	}

	return result
}

// 转换GBK编码到UTF8
func convertGBKToUTF8(text string) string {
	// 使用mahonia进行编码转换
	decoder := mahonia.NewDecoder("gbk")
	utf8Text := decoder.ConvertString(text)
	return utf8Text
}

// 移除字符串中的空字节
func removeNullBytes(s string) string {
	return strings.ReplaceAll(s, "\x00", "")
}

// 反转义消息
func unescapeMessage(data []byte) []byte {
	var buffer bytes.Buffer

	for i := 0; i < len(data); i++ {
		if i < len(data)-1 {
			switch {
			case data[i] == 0x5A && data[i+1] == 0x01:
				buffer.WriteByte(0x5B)
				i++
			case data[i] == 0x5A && data[i+1] == 0x02:
				buffer.WriteByte(0x5A)
				i++
			case data[i] == 0x5E && data[i+1] == 0x01:
				buffer.WriteByte(0x5D)
				i++
			case data[i] == 0x5E && data[i+1] == 0x02:
				buffer.WriteByte(0x5E)
				i++
			default:
				buffer.WriteByte(data[i])
			}
		} else {
			buffer.WriteByte(data[i])
		}
	}

	return buffer.Bytes()
}

// 计算CRC校验码
func calculateCRC(data []byte) uint16 {
	// 简化的CRC计算逻辑，实际项目中可能需要更复杂的算法
	var crc uint16 = 0
	for _, b := range data {
		crc += uint16(b)
	}
	return crc
}

// 转义消息 - 只用于反序列化时
func escapeMessage(data []byte) []byte {
	var buffer bytes.Buffer

	// 写入开始标识
	buffer.WriteByte(JT809_HEADER_FLAG_START)

	// 处理内部字节（跳过开始和结束标识）
	for i := 1; i < len(data)-1; i++ {
		switch data[i] {
		case 0x5B: // 0x5B
			buffer.WriteByte(JT809_ESCAPE_CHAR)
			buffer.WriteByte(JT809_ESCAPE_CHAR_5B)
		case 0x5A: // 0x5A
			buffer.WriteByte(JT809_ESCAPE_CHAR)
			buffer.WriteByte(JT809_ESCAPE_CHAR_5A)
		case 0x5D: // 0x5D
			buffer.WriteByte(JT809_ESCAPE_PREFIX_5D)
			buffer.WriteByte(JT809_ESCAPE_SUFFIX_5D)
		case 0x5E: // 0x5E
			buffer.WriteByte(JT809_ESCAPE_PREFIX_5E)
			buffer.WriteByte(JT809_ESCAPE_SUFFIX_5E)
		default:
			buffer.WriteByte(data[i])
		}
	}

	// 写入结束标识
	buffer.WriteByte(JT809_HEADER_FLAG_END)

	return buffer.Bytes()
}

// 序列化结构体为字节
func serializeStruct(data interface{}) ([]byte, error) {
	buf := new(bytes.Buffer)
	err := binary.Write(buf, binary.BigEndian, data)
	return buf.Bytes(), err
}

// 运行JT809通信服务
func RunJT809() {
	logger := g.Log()
	logger.Info(ctx, "正在启动JT809服务...")
	//g.Log().Error(ctx, "正在启动JT809服务...")
	// 创建客户端配置
	config := &JT809Config{
		ServerIP:          "47.111.114.204",
		ServerPort:        39698,
		HeartbeatInterval: 60,
		ReconnectInterval: 10,
	}

	// 创建客户端实例
	client := NewJT809Client(ctx, config)

	// 加载用户配置
	if err := client.LoadUserFromDB(); err != nil {
		logger.Fatal(ctx, "加载用户配置失败:", err)
		return
	}
	logger.Info(ctx, "成功加载用户配置")

	// 获取v1版本的登录消息（用于对比）
	getV1LoginMessage(ctx, logger)

	// 构建登录消息
	loginMsg, err := client.BuildLoginMessage()
	if err != nil {
		logger.Fatal(ctx, "构建登录消息失败:", err)
		return
	}
	logger.Info(ctx, "成功构建登录消息: ", hex.EncodeToString(loginMsg))

	// 解码心跳消息
	heartbeatBytes, err := hex.DecodeString(JT809_HEART_BEAT)
	if err != nil {
		logger.Fatal(ctx, "解码心跳消息失败:", err)
		return
	}
	logger.Info(ctx, "心跳消息: ", JT809_HEART_BEAT)

	// 连接重试计数
	reconnectCount := 0
	maxReconnectCount := 0 // 0表示无限重试

	// 主循环，更接近v1.Jt809的实现方式
	for {
		if maxReconnectCount > 0 && reconnectCount >= maxReconnectCount {
			logger.Error(ctx, "达到最大重连次数，停止重试")
			break
		}

		// 尝试建立连接
		serverAddr := fmt.Sprintf("%s:%d", config.ServerIP, config.ServerPort)
		logger.Info(ctx, "正在连接服务器:", serverAddr)
		conn, err := net.DialTimeout("tcp", serverAddr, JT809_CONN_TIMEOUT)
		if err != nil {
			reconnectCount++
			logger.Error(ctx, fmt.Sprintf("连接服务器失败(第%d次): %v", reconnectCount, err))
			time.Sleep(time.Duration(config.ReconnectInterval) * time.Second)
			continue
		}

		// 重置重连计数
		reconnectCount = 0
		logger.Info(ctx, "成功连接到服务器:", serverAddr)

		// 设置连接
		client.conn = conn

		// 发送登录消息
		logger.Info(ctx, "正在发送登录消息...")
		if err := client.Send(loginMsg); err != nil {
			logger.Error(ctx, "发送登录消息失败:", err)
			conn.Close()
			time.Sleep(time.Duration(config.ReconnectInterval) * time.Second)
			continue
		}
		logger.Info(ctx, "登录消息发送成功")

		// 启动心跳定时器 (但不使用gcron，直接在goroutine中处理)
		heartbeatStop := make(chan struct{})
		go func() {
			ticker := time.NewTicker(time.Duration(config.HeartbeatInterval) * time.Second)
			defer ticker.Stop()

			for {
				select {
				case <-ticker.C:
					logger.Debug(ctx, "正在发送心跳...")
					if err := client.Send(heartbeatBytes); err != nil {
						logger.Error(ctx, "发送心跳失败:", err)
						return
					}
					logger.Debug(ctx, "心跳发送成功:", gtime.Now().Format("Y-m-d H:i:s"))
				case <-heartbeatStop:
					logger.Debug(ctx, "心跳任务已停止")
					return
				}
			}
		}()

		// 启动消息处理队列
		processStop := make(chan struct{})
		go func() {
			for {
				select {
				case <-processStop:
					logger.Debug(ctx, "消息处理队列已停止")
					return
				default:
					value, ok := client.msgQueue.PopLeft()
					if ok && value != "" {
						processMessage(value)
					} else {
						time.Sleep(100 * time.Millisecond)
					}
				}
			}
		}()

		logger.Info(ctx, "开始接收消息")
		// 接收消息
		receiveBuffer := make([]byte, 4096)
		for {
			// 设置较长的读超时，避免永久阻塞但也不会过早断开
			if err := conn.SetReadDeadline(time.Now().Add(180 * time.Second)); err != nil {
				logger.Error(ctx, "设置读取超时失败:", err)
				break
			}

			n, err := conn.Read(receiveBuffer)
			if err != nil {
				if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
					// 读取超时，可能是网络暂时不可用，继续尝试
					logger.Warning(ctx, "读取超时，继续尝试...")
					continue
				}

				// 其他错误，断开连接并重连
				logger.Error(ctx, "接收消息错误:", err)
				break
			}

			if n > 0 {
				logger.Debug(ctx, "收到数据，长度:", n)

				// 复制数据，避免缓冲区复用导致的数据覆盖
				data := make([]byte, n)
				copy(data, receiveBuffer[:n])

				// 将二进制数据转为十六进制字符串
				hexStr := hex.EncodeToString(data)
				//logger.Info(ctx, "收到消息:", hexStr)

				// 推送到消息队列
				client.msgQueue.PushRight(hexStr)
			}
		}

		// 停止心跳和处理线程
		close(heartbeatStop)
		close(processStop)

		// 关闭连接
		conn.Close()

		logger.Info(ctx, "连接已断开，准备重新连接 - 当前时间:", gtime.Datetime())

		// 回收资源
		logger.Debug(ctx, "正在回收GC资源")
		runtime.GC()

		// 等待一段时间后重连
		time.Sleep(time.Duration(config.ReconnectInterval) * time.Second)
		logger.Info(ctx, "正在重新连接上级服务器")
	}
}
